package studio.raptor.extractor.segment.task.assemble;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.*;

import org.apache.http.client.HttpClient;
import studio.raptor.extractor.common.util.Pair;
import studio.raptor.extractor.segment.constant.JsonElementName;
import studio.raptor.extractor.segment.entity.EventDataInfo;
import studio.raptor.extractor.segment.entity.EventMeta;

/**
 * Created by liuyong8 on 2017/7/25.
 */
public class EventMetaMsgTask implements Callable<EventDataInfo> {
  private Map<String/* JsonElementName */, String/* serverUrl */> serverUrlMap;
  private EventMeta eventMeta;
  private ExecutorService eventMetaMsgExec;
  private HttpClient httpClient;

  private String[] nodeNames;

  public EventMetaMsgTask(Map<String, String> serverUrlMap, EventMeta eventMeta, ExecutorService eventMetaMsgExec,HttpClient httpClient) {
    this.serverUrlMap = serverUrlMap;
    this.eventMeta = eventMeta;
    this.eventMetaMsgExec = eventMetaMsgExec;
    this.httpClient = httpClient;
    List<String> nodeNameCcnfigs = new ArrayList<>();
    for (Entry<String,String> entry : serverUrlMap.entrySet()){
      nodeNameCcnfigs.add(entry.getKey());
    }
    this.nodeNames = nodeNameCcnfigs.toArray(new String[]{});
  }

  @Override
  public EventDataInfo call() throws Exception {
    CompletionService<Pair<String, String>> nodeMsgComp =
        new ExecutorCompletionService<Pair<String, String>>(eventMetaMsgExec);
    List<Pair<String,String>> nodes = new ArrayList<Pair<String,String>>();
    EventDataInfo eventDataInfo = new EventDataInfo();

    for (String nodeStr : nodeNames) {
      String serverUrl = serverUrlMap.get(nodeStr);
      NodeMsgTask nodeMsgTask = new NodeMsgTask(eventMeta, nodeStr, serverUrl,httpClient);
      nodeMsgComp.submit(nodeMsgTask);
    }

    for (int i=0;i<nodeNames.length;i++){
      Future<Pair<String, String>> future = nodeMsgComp.take();
      Pair<String, String> pair = future.get();
      nodes.add(pair);
    }
    eventDataInfo.setMetaId(eventMeta.getMetaId());
    eventDataInfo.setEventMeta(eventMeta);
    eventDataInfo.setNodes(nodes);
    return eventDataInfo;
  }



}
